package bigdata.monitor.binlog.utils;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.networknt.schema.JsonSchema;
import com.networknt.schema.JsonSchemaFactory;
import com.networknt.schema.SpecVersion;
import com.networknt.schema.ValidationMessage;
import lombok.extern.slf4j.Slf4j;

import java.util.Set;

/**
 * 使用 JSON Schema 对 Binlog 消息中的 "after" 字段进行基础规则校验。
 * 支持的规则语法（simple_condition 示例）：
 * - 类型校验：          "age:int", "price:double", "name:string"
 * - 范围校验（整数）：   "age:18-80"
 * - 枚举校验：          "status:1,2,3"
 * - 固定长度（字符串）： "phone:11"
 * - 正则校验：          "email:(^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$)"
 * - 宽松模式（仅数值）： "score:int?"  → 允许字符串形式的整数（如 "123"）
 * ⚠️ 注意：
 * - 本评估器 **强制要求规则所涉及的字段必须存在于 "after" 中**。
 * 如果字段不存在，则视为校验失败（因为规则意图是校验该字段，缺失即违规）。
 * - 字符串类型不支持宽松模式（"string?" 会抛出异常）。
 */
@Slf4j
public class JsonSchemaRuleEvaluator {

    /**
     * 全局共享的 Jackson ObjectMapper 实例，用于 JSON 解析与构建。
     */
    private static final ObjectMapper mapper = new ObjectMapper();
    /**
     * JSON Schema 工厂实例，使用 Draft 7 规范（兼容 networknt/json-schema-validator）。
     */
    private static final JsonSchemaFactory factory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V7);

    /**
     * 对给定的 Binlog JSON 消息和简单规则条件执行校验。
     *
     * @param binlogJson      Flink CDC 输出的完整 Binlog JSON 字符串，格式如：
     *                        {"table":"user","op":"u","before":{...},"after":{"age":150,"name":"Alice"}}
     * @param simpleCondition 规则条件字符串，格式为 "字段名:规则"，例如 "age:int" 或 "age:18-80"
     * @return true 表示符合规则；false 表示违规或校验过程中出错（fail-safe）
     */
    public static boolean evaluate(String binlogJson, String simpleCondition) {
        try {
            // 提取 "after" 节点作为校验对象（若 after 为空或缺失，则用空对象代替）
            JsonNode after = extractAfter(binlogJson);
            // 根据 simpleCondition 动态构建 JSON Schema 字符串
            String schemaJson = buildSchema(simpleCondition);
            // 将 schema 字符串解析为 JsonNode
            JsonNode schemaNode = mapper.readTree(schemaJson);
            // 编译 JSON Schema
            JsonSchema schema = factory.getSchema(schemaNode);
            // 执行校验
            Set<ValidationMessage> errors = schema.validate(after);
            // 无错误即通过
            return errors.isEmpty();
        } catch (Exception e) {
            // 捕获所有异常（包括规则语法错误、JSON 解析失败等），记录日志并返回 false（安全失败）
            log.warn("JSON Schema validation failed. condition: {}", simpleCondition, e);
            return false; // 安全失败
        }
    }

    /**
     * 从 Binlog JSON 中提取 "after" 字段对应的 JsonNode。
     *
     * @param binlogJson 完整的 Binlog JSON 字符串
     * @return "after" 节点；如果缺失或为 null，则返回一个空的 ObjectNode（避免后续空指针）
     * @throws Exception 若 JSON 解析失败
     */
    private static JsonNode extractAfter(String binlogJson) throws Exception {
        JsonNode root = mapper.readTree(binlogJson);
        JsonNode after = root.path("after");
        // 如果 after 不存在或为 null，返回空对象（这样 required 校验会失败，符合预期）
        return after.isMissingNode() || after.isNull() ? mapper.createObjectNode() : after;
    }

    /**
     * 根据 simpleCondition 字符串动态构建 JSON Schema。
     * 格式：field:spec
     * 例如：age:int, status:1,2,3, phone:11, email:(regex)
     *
     * @param simpleCondition 规则字符串
     * @return 符合 JSON Schema Draft 7 规范的 schema 字符串
     * @throws IllegalArgumentException 当规则格式不合法时抛出
     */
    private static String buildSchema(String simpleCondition) {
        if (simpleCondition == null || !simpleCondition.contains(":")) {
            throw new IllegalArgumentException("Invalid simple_condition: " + simpleCondition);
        }

        // 拆分字段名和规则描述
        String[] parts = simpleCondition.split(":", 2);
        String field = parts[0].trim();
        String spec = parts[1].trim();

        // 判断是否为宽松模式（以 ? 结尾）
        boolean lenient = spec.endsWith("?");
        String typeOrValue = lenient ? spec.substring(0, spec.length() - 1) : spec;

        // 创建顶层 schema 对象：类型为 object
        ObjectNode schema = mapper.createObjectNode();
        schema.put("type", "object");

        // 创建 properties 节点
        ObjectNode props = schema.putObject("properties");
        // 为目标字段创建 schema
        ObjectNode fieldSchema = props.putObject(field);
        // 强制要求该字段必须存在
        // 任何规则要校验的字段，都必须出现在 "after" 中，否则视为违规
        schema.putArray("required").add(field);

        // 根据规则类型构建具体校验逻辑
        if (typeOrValue.equalsIgnoreCase("int") || typeOrValue.equalsIgnoreCase("integer")) {
            if (lenient) {
                buildLenientInteger(fieldSchema);
            } else {
                fieldSchema.put("type", "integer");
            }
        } else if (typeOrValue.equalsIgnoreCase("double") || typeOrValue.equalsIgnoreCase("float")) {
            if (lenient) {
                buildLenientNumber(fieldSchema);
            } else {
                fieldSchema.put("type", "number");
            }
        } else if (typeOrValue.equalsIgnoreCase("string")) {
            if (lenient) {
                // 字符串不支持宽松模式（因为 string 本身已是最宽泛类型）
                throw new IllegalArgumentException("String does not support lenient mode");
            }
            fieldSchema.put("type", "string");
        } else if (spec.contains("-") && spec.matches("^\\d+(\\.\\d+)?-\\d+(\\.\\d+)?$")) {
            // 支持整数或小数范围，例如：18-80, 0.0-5.0, 0-100.5
            String[] bounds = spec.split("-", 2); // 只分割一次，防止负数干扰（虽然你业务可能无负数）
            String minStr = bounds[0].trim();
            String maxStr = bounds[1].trim();

            try {
                // 尝试解析为 double（兼容整数和小数）
                double min = Double.parseDouble(minStr);
                double max = Double.parseDouble(maxStr);

                // 检查原始字符串是否包含小数点
                boolean hasDecimalPoint = minStr.contains(".") || maxStr.contains(".");

                if (!hasDecimalPoint) {
                    // 如果两边都没有小数点，才视为整数范围
                    fieldSchema.put("type", "integer")
                            .put("minimum", (int) min)
                            .put("maximum", (int) max);
                } else {
                    // 只要任意一边写了 .，就视为小数范围
                    fieldSchema.put("type", "number")
                            .put("minimum", min)
                            .put("maximum", max);
                }
            } catch (NumberFormatException e) {
                throw new IllegalArgumentException("Invalid range format: " + spec);
            }
        } else if (spec.contains(",")) {
            // 枚举校验：如 "1,2,3" → 转为字符串枚举（JSON Schema enum 要求 string）
            buildEnum(fieldSchema, spec.split(","));
        } else if (spec.startsWith("(") && spec.endsWith(")")) {
            // 正则校验：如 "(^[0-9]{11}$)"
            String regex = spec.substring(1, spec.length() - 1); // 去掉括号
            return buildPatternSchema(field, regex);
        } else {
            // 尝试解析为固定长度（整数）
            try {
                int len = Integer.parseInt(spec);
                fieldSchema.put("type", "string").put("minLength", len).put("maxLength", len);
            } catch (NumberFormatException e) {
                throw new IllegalArgumentException("Unknown spec: " + spec);
            }
        }
        // 返回 schema 的 JSON 字符串表示
        return schema.toString();
    }

    /**
     * 构建宽松整数校验：允许整数或符合整数格式的字符串。
     * 生成的 schema：
     * {
     *   "anyOf": [
     *     {"type": "integer"},
     *     {"type": "string", "pattern": "^-?(0|[1-9][0-9]*)$"}
     *   ]
     * }
     * @param fieldSchema 目标字段的 schema 节点
     */
    private static void buildLenientInteger(ObjectNode fieldSchema) {
        fieldSchema.putArray("anyOf")
                .add(mapper.createObjectNode().put("type", "integer"))
                .add(mapper.createObjectNode()
                        .put("type", "string")
                        .put("pattern", "^-?(0|[1-9][0-9]*)$"));
    }

    /**
     * 构建宽松数值校验：允许 number 或符合浮点数格式的字符串。
     * 支持科学计数法（如 1.23e5）
     *
     * @param fieldSchema 目标字段的 schema 节点
     */
    private static void buildLenientNumber(ObjectNode fieldSchema) {
        fieldSchema.putArray("anyOf")
                .add(mapper.createObjectNode().put("type", "number"))
                .add(mapper.createObjectNode()
                        .put("type", "string")
                        .put("pattern", "^-?(0|[1-9][0-9]*)(\\.[0-9]+)?([eE][-+]?[0-9]+)?$"));
    }

    /**
     * 构建枚举校验：将逗号分隔的值列表转换为 JSON Schema 的 enum。
     * 注意：JSON Schema 的 enum 要求是字符串，所以即使输入是数字（如 "1,2"），
     * 也会被当作字符串处理。这与 CDC 中字段可能是字符串类型一致。
     *
     * @param fieldSchema 目标字段的 schema 节点
     * @param values      枚举值字符串数组（如 ["1", "2", "3"]）
     */
    private static void buildEnum(ObjectNode fieldSchema, String[] values) {
        fieldSchema.put("type", "string");
        ArrayNode enumArray = fieldSchema.putArray("enum");
        for (String v : values) {
            enumArray.add(v.trim());
        }
    }

    private static String buildPatternSchema(String field, String regex) {
        ObjectNode schema = mapper.createObjectNode();
        schema.put("type", "object");
        ObjectNode props = schema.putObject("properties");
        ObjectNode fieldSchema = props.putObject(field);
        fieldSchema.put("type", "string")
                .put("pattern", regex); // JSON Schema 原生支持
        // 同样需要 required
        schema.putArray("required").add(field);
        return schema.toString();
    }
}
